第 2 课：LLM 作为决策引擎（Decision Engine）

你需要掌握：

1.1 如何让 LLM 做“推理 + 决策”
1.2 Prompt 架构（System Prompt、Role Prompt、Loop Prompt）
1.3 如何让 LLM 可控（避免幻觉）
1.4 LLM 的三种模式：
- 纯文本聊天模型（OpenAI GPT、Llama）
- 工具调用模型（function-calling 模式）
- 思维链模型（CoT）
1.5 决策类 Prompt 的结构化格式：

示例（工程化格式）：

Thought:
Action:
Action Input:

1.6 如何让模型“只输出 JSON”
1.7 如何限制模型不跑偏（Guardrails + Validators）
1.8 推理深度控制（max_steps、deliberate_thinking）

（一）如何让 LLM 做“推理 + 决策”

在 Agent 系统里，LLM 不再是“回答问题的聊天机器人”，而是一个 Policy（策略函数）：

输入：当前状态 state（含目标、历史、工具可用性、约束）输出：下一步动作 action（包括 finish / tool_call / ask_user / replan）

后面所有模块（ReAct / Planner / State Machine）都建立在这个认知之上。

工程上，你必须把“决策”变成可解析的结构化输出，而不是自然语言随便说。

决策输出的两种主流形态

形态 A：Action JSON（推荐） 适合工程解析、状态机、工具路由
形态 B：Thought/Action/Obs（教学友好） 适合调试，但不如 JSON 稳

同时，你还要定义这个 agent 可以做的所有决策和动作，这称为 决策空间

必须把动作空间写进 system prompt，是可控性的第一步。

典型 Agent 最小动作集合：

finish：输出最终回答

tool_call：调用某个工具

ask_user：信息不足，向用户提问

replan：当前策略失败，重做计划

（二）Prompt 架构

工程上建议你把 prompt 分成三层（System / Role / Loop），并且代码里明确分离，便于维护、A/B test、版本化。

核心目的只有三点：

可控性：System 里放“硬规则”，避免模型跑偏。
可维护性：Role 变化频繁；Loop 每轮变化更频繁。分层能减少改动面。
可实验性：A/B test 只替换某一层（通常是 Role 或 System 的某个片段），其余不动，才能对比出“因果效果”。

1. System Prompt

System 是“宪法”，原则是：只放长期稳定且强约束的内容。

包含：

允许的动作集合（action space）
输出必须为 JSON（schema）
禁止输出推理过程（如果你要）
失败时的 fallback（例如：JSON 解析失败如何纠正）
工具使用原则（“没有 observation 不得引用工具结果”）

建议你把 System 写成“规格文档式”的 Prompt：

明确 schema
明确 allowed values
明确 error policy

示例（决策引擎 System）：

You are an agent decision engine.

Output MUST be exactly one valid JSON object. No markdown, no extra text.

Allowed actions: ["finish","tool_call","ask_user","replan"].

Schema:
{
  "action": "...",
  "tool": "string|null",
  "tool_input": "object|null",
  "final": "string|null"
}

Rules:
- Never claim tool results without an Observation.
- If output is not valid JSON, self-correct and output valid JSON only.
- Be concise.

2. Role Prompt

Role 是“岗位说明书”（可变：任务角色与偏好），随场景变化。放：

任务类型（网页调研 / 代码生成 / 竞品分析 / 运营文案等）
风格偏好（简短、bullet、严格引用等）
质量标准（必须给出可执行步骤、必须给出风险提示等）

Role 的关键：同一系统规则下，切换岗位能力。

示例：

“你是网页调研 agent，必须给出来源链接/引用”
“你是代码审查 agent，优先指出安全风险”
“你是数据分析 agent，输出要包含可复现步骤”

例如：

“你是一个网页调研 agent”
“你输出要简短”
“你必须引用来源（如果有 web 工具）”

3. Loop Prompt（每轮动态：状态输入）

Loop 是“当下工单”，每轮都变。放：

当前目标（goal）
可用工具清单（tools + schema）
关键历史 / 摘要（memory summary）
预算（max_steps / token / cost）
上一次工具 observation / error（用于纠错与 replanning）

Loop 的关键：把 agent 的“状态”显式喂给模型，让它基于状态做决策，而不是凭空发挥。

Loop 内容建议统一用 JSON（对模型和你都更清晰，方便日志与回放）：

{
  "goal": "...",
  "tools_available": [...],
  "memory_summary": "...",
  "budget": {"max_steps": 6, "max_tool_calls": 3},
  "last_observation": "...",
  "last_error": null
}

例子

import os, json, requests
from dataclasses import dataclass

API_KEY = os.getenv("COMET_API_KEY")
URL = "https://api.cometapi.com/v1/chat/completions"
if not API_KEY:
    raise RuntimeError("COMET_API_KEY not set")

SYSTEM_PROMPT = """
You are an agent decision engine.

Output MUST be exactly one valid JSON object. No markdown, no extra text.

Allowed actions: ["finish","tool_call","ask_user","replan"].

Schema:
{
  "action": "finish|tool_call|ask_user|replan",
  "tool": "string|null",
  "tool_input": "object|null",
  "final": "string|null"
}

Rules:
- Never claim tool results without an Observation.
- If you need external info, use tool_call.
- If information is missing and no tool can help, use ask_user with a clear question in final.
- Keep final short.
""".strip()

role_prompt = """
    You are a web research agent.
    Prefer bullet points.
    If you propose using a tool, specify the tool name and tool_input clearly.
    """.strip()

loop_state = {
    "goal": "Explain what an agent is with one concrete example.",
    "tools_available": ["search", "http_get"],
    "memory_summary": "User prefers concise bullet points.",
    "budget": {"max_steps": 6, "max_tool_calls": 3},
    "last_observation": None,
    "last_error": None
}


@dataclass
class PromptPack:
    system: str
    role: str
    loop: dict

def call_llm(prompt_pack: PromptPack, model="gpt-4o", max_tokens=220, temperature=0.2):
    messages = [ # 解包传入的提示词结构
        {"role": "system", "content": prompt_pack.system},
        {"role": "system", "content": f"ROLE:\n{prompt_pack.role}".strip()},
        {"role": "user", "content": json.dumps(prompt_pack.loop, ensure_ascii=False)}
    ]
    print(type(json.dumps(prompt_pack.loop, ensure_ascii=False)))
    headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
    payload = {"model": model, "messages": messages, "max_tokens": max_tokens, "temperature": temperature}

    # 请求
    r = requests.post(URL, headers=headers, json=payload, timeout=30)
    r.raise_for_status()
    print(type(r))       # <class 'requests.models.Response'>
    text = r.json()["choices"][0]["message"]["content"]  # 格式化
    print(type(text))    # str 
    result = json.loads(text)  # 这一步确保str 可以转化成 JSON 格式，否则会报错
    print(type(result))  # dict  

    return result

if __name__ == "__main__":
    pack = PromptPack(system=SYSTEM_PROMPT, role=role_prompt, loop=loop_state)
    decision = call_llm(pack)
    
    print_json(decision)

输出如下：

{
  "action": "tool_call",
  "tool": "search",
  "tool_input": {
    "query": "definition of an agent with example"
  },
  "final": null
}

思考两个问题

@dataclass PromptPack 到底是什么？为什么不用原生 dict？

这段代码定义了一个对象（class instance），用内部的三个属性的值来代表具体的 prompt

相比于 dict，对象的优点在于：
- 可以规定属性数据类型
- 更安全
- 更容易修改，包括属性名和内容
- 更容易拓展和追加新的字段
```
@dataclass
class PromptPack:
    system: str
    role: str
    loop: dict
    version: str
    experiment: str | None = None
```
- “Prompt ≠ JSON”: PromptPack 是你本地的“工程对象”，而不是你发给 LLM 的 payload。你在 call_llm() 里才把它“展开”为 messages
为什么有的地方用了 .strip()，有的地方没有

.strip() 是为了消除去掉 开头和结尾 的空格，\n, \t，这是因为三引号字符串天然会带一个“首尾空行”

例如
```
"""
hello
"""
```
实际内容是： \nhello\n

某些模型对 prompt 的第一行很敏感（尤其 system prompt）。去掉前导空白是个好习惯。

总之：对“展示给模型看的自然语言 Prompt” → 可以 strip，而对“结构化数据 / JSON / state” → 不要 strip

其他知识

A/B test 是什么

A/B 测试就是：在控制其他变量不变的前提下，只改变一个因素（例如 Role Prompt 的一段话），比较产出指标的差异，从而判断哪个更好。

你可以用同一批 loop_state，跑两套 role prompt：
ROLE_A = "You are a web research agent. Be concise."
ROLE_B = "You are a web research agent. Be concise. Never fabricate sources."

# 对同一批 tasks 分别 call_llm，统计 JSON 合法率、action 分布、是否触发 ask_user 等

如何版本化 Prompt:

Prompt 版本化的目标：

任何线上结果可追溯到“当时用的 Prompt”

发生质量回退可快速回滚

支持 A/B test 与灰度发布

版本化的三种粒度（从轻到重）

字符串版本号（最简单） SYSTEM_PROMPT_VERSION="sys_v1.3.0"

文件版本化（Git）（推荐） prompt 存到 repo，以文件差异追踪

Prompt Registry（大系统）数据库/配置中心存 prompt，带发布流程

推荐的目录结构
prompts/
  system/
    decision_engine_v1.0.txt
    decision_engine_v1.1.txt
  roles/
    web_research_v1.0.txt
    coder_v1.0.txt
  loop_templates/
    state_schema_v1.0.json

(三）如何避免错误

大模型输出是不稳定的，那么我们如何确保模型能按照我们要求的内容和格式输出，而不是胡言乱语，并在出现问题时能自动诊断呢？

我们一般会采取三个步骤

输出结构约束：只允许 JSON schema
动作空间约束：只能从给定 action 里选
后验验证：解析失败就重试 / 纠错 / 降级

1. 约束(Guardrails)

import os
import json
import time
import requests
from typing import Any, Dict, Optional, Tuple

设定动作空间和工具白名单

ALLOWED_ACTIONS = {"finish", "tool_call", "ask_user", "replan"} # 动作空间约束
ALLOWED_TOOLS = {"search"}  # 工具空间

# 虚拟定义几个工具
def tool_search(query: str) -> str:
    return f"网络结果显示: Agent = LLM+tools+memory loop."

TOOLS = {"search": tool_search} # 把 ALLOWED_TOOLS 中的 工具名 和 具体的工具函数对应起来

设定 system 提示词模板 (工具类常用)

SYSTEM_PROMPT = f"""
You are an agent decision engine.

You MUST output exactly one JSON object. No markdown, no extra text.

Allowed actions: {sorted(ALLOWED_ACTIONS)}.
Allowed tools (only if action is tool_call): {sorted(ALLOWED_TOOLS)}.

Schema:
{{
  "action": "finish|tool_call|ask_user|replan",
  "tool": "string|null",
  "tool_input": "object|null",
  "final": "string|null"
}}

Rules (anti-hallucination):
- Tool results can ONLY come from an Observation provided by the system. Never invent search results.
- If you need external info, choose tool_call and wait for Observation.
- If the required tool is not available, choose ask_user or replan.
- If you reference any tool result in final, you must have received it in Observation in the current conversation.
""".strip()

单次调用函数

def call_llm(messages, model="gpt-4o", max_tokens=220, temperature=0.2) -> str: # 规定输出只能是 str
    headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
    payload = {
        "model": model,
        "messages": messages,
        "max_tokens": max_tokens,
        "temperature": temperature,
    }
    r = requests.post(CHAT_URL, headers=headers, json=payload, timeout=30)
    r.raise_for_status()
    return r.json()["choices"][0]["message"]["content"]

2. 验证(Validators)

就算我们在 prompt 中限制了模型的行为规范，模型仍有概率不按照要求进行，因此我们要面临以下问题：

模型输出不是一个严格的 json 格式

使用了不在工具列表中的工具

执行的 action 和使用的 tool 不匹配
多次尝试仍然反复错误，如何及时止损

没有 observation 却引用“搜索结果”

输出结果规范性验证函数

回顾之前定义的回复格式

{{
  "action": "finish|tool_call|ask_user|replan",
  "tool": "string|null",
  "tool_input": "object|null",
  "final": "string|null"
}}

因此检查要包含以下内容：

输出结果中是否缺失 System Prompt 模板中的任何字段
输出结果的“Action” 中是否包含 System Prompt 模板外的任何字段
（输出结果的“Action" 为 "tool_call" 时），输出结果的”tool“ 是否合法
- 明明Action" 为 "tool_call"，却没有调用 tool
- 调用的 tool 不在允许范围中
（输出结果的“Action" 为 "tool_call" 时），输出结果的”tool_input“ 是否为 dict
（输出结果的“Action" 为 "tool_call" 时），输出结果的”final“ 是否为 null
（输出结果的“Action" 非 "tool_call" 时），输出结果的”tool/tool_input“ 是否为 null
（输出结果的“Action" 非 "tool_call" 时），输出结果的”final“ 是否为 str

def validate_decision(obj: Dict[str, Any], tools_available: set) -> Tuple[bool, str]:  #规定输出是元组
    """后验验证：schema + action/tool 白名单 + 基本一致性检查"""
    
    # 检查一：输出结果中是否缺失 System Prompt 模板中的任何字段
    for k in ("action", "tool", "tool_input", "final"):
        if k not in obj:
            return False, f"Missing key: {k}"

    # 检查二：输出结果的“Action” 中是否包含 System Prompt 模板外的任何字段
    action = obj["action"]
    if action not in ALLOWED_ACTIONS:
        return False, f"Invalid action: {action}"

    # 检查三：（输出结果的“Action" 为 "tool_call" 时），输出结果的”tool“ 是否合法
    if action == "tool_call":
        tool = obj["tool"]
        # 检查 3.1： 明明Action" 为 "tool_call"， 却没有 调用 tool
        if not isinstance(tool, str):
            return False, f"Tool not allowed/available: {tool}"
        # 检查 3.2： 调用的 tool 不在允许范围中
        if tool not in tools_available:
            return False, f"Tool not allowed/available: {tool}"
        
    # 检查四：（输出结果的“Action" 为 "tool_call" 时），输出结果的”tool_input“ 是否为 dict
    if action == "tool_call":
        if not isinstance(obj["tool_input"], dict):
            return False, "tool_input must be an object when action=tool_call"
        
    # 检查五：（输出结果的“Action" 为 "tool_call" 时），输出结果的”final“ 是否为 null
    if action == "tool_call":
        if obj["final"] is not None:
            return False, "final must be null when action=tool_call"
    
    # 检查六：（输出结果的“Action" 非 "tool_call" 时），输出结果的”tool/tool_input“ 是否为 null
    else:
        if obj["tool"] is not None or obj["tool_input"] is not None:
            return False, "tool and tool_input must be null when action is not tool_call"
        
    # 检查七：（输出结果的“Action" 非 "tool_call" 时），输出结果的”final“ 是否为 str
        if not isinstance(obj["final"], str):
            return False, "final must be a string when action is not tool_call"

    return True, "ok"

isinstance(object, classinfo):

object：要检查的对象

classinfo：类型（如 str, int, dict），或类型元组（如 (str, int)）

返回：

True 或 False

错误纠正和重复尝试

def decide_with_retry(
    state: Dict[str, Any],   # 当前目标
    tools_available: set,    # 可用工具名 set
    last_observation: Optional[str] = None, # 上一轮 Agent 执行 Action 后，获得的工具执行结果，比如搜索函数的返回结果
    max_retries: int = 2,    # 最大允许尝试次数
    ) -> Dict[str, Any]:     
    

    # 基础信息： 系统提示词模板 + 初始目标 + 可用工具名
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": json.dumps({"state": state, "tools_available": sorted(tools_available)}, ensure_ascii=False)},
    ]

    # 如果不是第一次对话，就会有之前的工具调用结果，将其添加到对话中
    if last_observation is not None:
        # 注意：Observation 用 system role 注入，明确告诉模型“这是唯一可信工具结果来源”
        messages.append({"role": "system", "content": f"Observation:\n{last_observation}"})

    # 错误检查 和 循环尝试
    err_msg = None
    for attempt in range(max_retries + 1):  # 在有限的循环次数内，添加错误信息并尝试

        # 如果上次发送得到的回答 经过检查存在错误，本次对话中加上上次的报错信息。
        if err_msg:
            messages.append({
                "role": "system",
                "content": f"Your previous output failed validation: {err_msg}. "
                           f"Output ONLY one valid JSON object that matches the schema."
            })

        # 获取回答
        raw = call_llm(messages) 
        # 注意这里获取的 raw 是已经从原始json中提取出的 回答部分 str，但是可以 json 化

        # 错误类型一：回答内容str 不能严格按照 json 格式
        try:
            obj = json.loads(raw) # 把 str 转化成 dict
        except Exception as e:
            err_msg = f"Invalid JSON parse error: {type(e).__name__}"
            continue

        # 错误类型二：回答内容存在逻辑错误，使用之前定义的函数检测
        ok, reason = validate_decision(obj, tools_available)
        if ok: # 如果成功，则返回回答的结果 dict。包含action，tool，tool_input， final
            return obj
        else: # 如果失败，添加错误原因到对话中
            err_msg = reason

        # 指数退避（简单版）
        time.sleep(0.4 * (2 ** attempt))

    # 降级策略：多次失败后，输出 ask_user（或 finish 的保守回答）
    return {
        "action": "ask_user",
        "tool": None,
        "tool_input": None,
        "final": "I couldn't produce a valid tool/action plan. What exactly do you want to achieve, and what constraints should I follow?"
    }

新数据类型：set

set 是 Python 的一种内置数据类型，表示一个**“无序、去重的元素集合”**。

例如
s = {"finish", "tool_call", "ask_user"}
set 的核心特性（这 4 个最重要）

自动去重
s = {"a", "b", "a"}
print(s)   # {'a', 'b'}
无序（不能用索引）
s = {"a", "b", "c"}
s[0]   # ❌ TypeError
成员判断极快，比 list 快三个数量级
"tool_call" in s   # 非常快
支持集合运算
A = {"finish", "tool_call"}
B = {"tool_call", "ask_user"}

A | B   # 并集 {'finish','tool_call','ask_user'}
A & B   # 交集 {'tool_call'}
A - B   # 差集 {'finish'}
set 的常见创建方式
# 字面量
s = {"a", "b", "c"}

# 从 list 转
s = set(["a", "b", "a"])

# 空 set（注意！）
s = set()    # ✅
s = {}       # ❌ 这是 dict

什么是 指数退避？

指数退避 = 出错后不要立刻重试，而是“越失败，等得越久”，以避免系统失控或被封禁。

time.sleep(0.4 * (2 ** attempt))

假设 attempt 从 0 开始：

attempt	sleep 时间
0	0.4 × 1 = 0.4s
1	0.4 × 2 = 0.8s
2	0.4 × 4 = 1.6s
3	0.4 × 8 = 3.2s

每失败一次，等待时间翻倍，以防止API 被 ban ，网络抖动之类的问题，进而导致 Agent 卡死在一个任务上，整个系统吞吐掉到 0

执行流程

流程为：首次对话决定是否需要工具 -> AI 认为当前资料不够回答该问题，需要调用工具->提取回答中需要使用的工具名，并去调用，获得返回 Observation-> 把 Observation放入第二次聊天prompt中

goal = "Explain what an agent is. If you need external info, use search."

tools_available = set(TOOLS.keys()) # 可用工具 set
state = {"goal": goal} 

# 第一步：让模型决定要不要用工具
decision1 = decide_with_retry(state, tools_available)
print("Decision1:", json.dumps(decision1, ensure_ascii=False))

if decision1["action"] == "tool_call":  # 如果第一次得到的回答是”要用工具
    tool = decision1["tool"]            # 获取工具的名字
    tool_input = decision1["tool_input"]# 获取工具需要的输入

    # 获取工具执行的 返回（也就是 obeservation）
    try:
        if tool == "search":
            query = tool_input.get("query", "")
            obs = TOOLS["search"](query)  # 执行工具
        else:
            raise RuntimeError("Unknown tool")
    except Exception as e:
        obs = f"[TOOL_ERROR] {type(e).__name__}: {e}"


    # 第二步：把 Observation 注入，要求模型基于 observation 输出 final
    decision2 = decide_with_retry(
        state={"goal": goal, "note": "Use the observation to answer."}, #注明使用 observation 来回答
        tools_available=tools_available,
        last_observation=obs
    )
    print("Decision2:", json.dumps(decision2, ensure_ascii=False))
    
else:
    # 不需要工具，直接 final
    print("Final:", decision1["final"])

Decision1: {"action": "tool_call", "tool": "search", "tool_input": {"query": "definition of an agent"}, "final": null}
Decision2: {"action": "finish", "tool": null, "tool_input": null, "final": "An agent is a system that combines a language model (LLM) with tools and a memory loop to perform tasks and make decisions."}

易混淆的工具

json.dumps(Dictionary, ensure_ascii=False): 把 dict 转成 str
json.loads(text): 把 str 转化成 dict
responce.json(): 把 'requests.models.Response' 转成 dict

（四）三种模式

工程上你要知道何时用哪一种：

纯文本（最通用）

适合：摘要、解释、写作
风险：不可控、难解析

function-calling（工具调用模式）

适合：稳定路由、严格参数、工具执行
优点：可解析、可验证、可重试
你后续模块 6 会深入

CoT（思维链）

适合：复杂推理、规划、数学
风险：冗长、成本高、可能跑偏
工程上通常用“短推理 + 验证”，或用“隐式推理，只输出结论+结构”

这些在后面几节课中会更详细地介绍。

（一）如何让 LLM 做“推理 + 决策”​

（二）Prompt 架构​

1. System Prompt​

2. Role Prompt​

3. Loop Prompt（每轮动态：状态输入）​

其他知识​

(三）如何避免错误​

1. 约束(Guardrails)​

设定动作空间和 工具白名单​

设定 system 提示词模板 (工具类常用)​

单次调用函数​

2. 验证(Validators)​

输出结果规范性验证函数​

错误纠正 和 重复尝试​

执行流程​

（四）三种模式​

纯文本（最通用）​

function-calling（工具调用模式）​

CoT（思维链）​